<?php
/**
 * Created by PhpStorm.
 * Script Name: Follow.php
 * Create: 2022/10/31 8:32
 * Description:
 * Author: fudaoji<fdj@kuryun.cn>
 */

namespace app\mp\event;


use app\common\coroutine\Client;
use app\common\event\Base;
use app\common\model\MpTemplatemsgTask;
use think\facade\Log;
use Swoole\Timer;

class Templatemsg extends Base
{
    /**
     * @var MpTemplatemsgTask
     */
    private $taskM;

    public function __construct()
    {
        parent::__construct();
        $this->taskM = new MpTemplatemsgTask();
        ini_set('memory_limit', '-1');
        set_time_limit(0);
    }

    public function timer($params = []){
        $job = $params['job'];
        echo '任务执行了' . $job->attempts() . '次' . PHP_EOL;
        $count = 0;
        \Swoole\Timer::after(3000, [$this, 'timerCallback'], $count);
        echo '续约：' . microtime(true) . PHP_EOL;
        /*controller('common/TaskQueue', 'event')->push([
            'delay' => 5,
            'params' => [
                'do' => ['\\app\\mp\\event\\Templatemsg', 'timer'],
                'delay' => 5,
            ]
        ]);*/
    }

    public function timerCallback(&$count){
        $count++;
        echo $count.':' . microtime(true) . PHP_EOL;
        if($count < 100){
            \Swoole\Timer::after(3000, [$this, 'timerCallback'], $count);
        }
    }

    /**
     * 测试
     * @param $params
     * @return array|bool|false|mixed|\PDOStatement|string|\think\Model
     * @throws \think\exception\DbException
     * @throws \think\Exception
     */
    public function testTask($params = []){
        $job = $params['job'];
        $task_id = $params['task_id'];
        $limit = $params['limit'];

        $task = $this->taskM->getOneJoin([
            'alias' => 'task',
            'join' => [
                ['mp_templatemsg msg', 'msg.id=task.templatemsg_id']
            ],
            'where' => ['task.id' => $params['task_id']/*, 'task.status' => ['in', [0, 1]], 'push_time' => ['<=', time()]*/],
            'field' => ['task.*', 'msg.template_id', 'msg.content', 'msg.url']
        ]);

        echo "task:" . $task_id . PHP_EOL;
        $redis = $this->getRedis();
        if(empty($task) || $job->attempts() > 10000){
            $job->delete();
            $redis->del('tempmsg_count_'.$task_id);
            echo('任务不存在');
            return true;
        }

        $rkey = 'openidlist' . $task_id;
        $refresh = !$redis->get('taskon'.$task_id);
        $mp_info = model('common/mp')->getOne($task['mpid']);
        $app = controller('mp/mp', 'event')->getApp($mp_info);

        if($refresh){ //首次执行
            $redis->del($rkey);
            $redis->set('taskon'.$task_id, 1);
            //生产
            $next_openid = '';
            while (true) {
                if($redis->lLen($rkey) >= $limit){
                    break;
                }
                $res = $app->user->list($next_openid);
                if (!empty($res['count'])) {
                    $openid_list = $res['data']['openid'];
                    $redis->lPush($rkey, ...$openid_list);
                    $next_openid = $res['next_openid'];
                }else{
                    break;
                }
            }
        }

        $config = config('system.develop');
        $coroutine = null;/*new Client([
            'port' => empty($config['swoole_port']) ? 9502 : $config['swoole_port'],
            'host' => empty($config['swoole_host']) ? '0.0.0.0' : $config['swoole_host']
        ]);*/
        echo('长度:' . $redis->lLen($rkey)) . PHP_EOL;
        $params = ['config' => $config, 'mp_info' => $mp_info, 'redis'=>$redis, 'rkey' => $rkey,
            'task' => $task, 'coroutine' => $coroutine
        ];

        swoole_timer_after(5000, [$this, 'afterCallback'], $params);
        if($redis->lLen($rkey)){
            //$job->release(1); //续约任务
        }else{
            $job->delete();
            $redis->del('tempmsg_count_'.$task_id);
        }
    }

    public function afterCallback($params = []){
        $coroutine = $params['coroutine'];
        $mp_info = $params['mp_info'];
        $app = controller('mp/mp', 'event')->getApp($mp_info);
        $redis = $params['redis'];
        $rkey = $params['rkey'];
        $task = $params['task'];
        $step = $params['step'] ?? 30;
        $interval = $params['interval'] ?? 5000;

        unset($openid_list);
        $openid_list = [];
        $count = 0;
        $api = $app->user;
        while ($count < $step && $d = $redis->lPop($rkey)){
            $count++;
            var_dump($api->get($d));
        }
        echo ('执行' . $count . "次<br>" . PHP_EOL);

        if($redis->lLen($rkey)){
            gc_mem_caches(); //清理缓存碎片
            echo "剩余任务量：" . $redis->lLen($rkey) . PHP_EOL;
            swoole_timer_after($interval, [$this, 'afterCallback'], $params); //持续
        }else{
            $redis->del('taskon'.$task['id']);
            echo "完成" . PHP_EOL;
        }
    }
    public function afterCallback1($params = []){
        $coroutine = $params['coroutine'];
        $mp_info = $params['mp_info'];
        $redis = $params['redis'];
        $rkey = $params['rkey'];
        $task = $params['task'];
        $step = $params['step'] ?? 30;
        $interval = $params['interval'] ?? 5000;

        unset($openid_list);
        $openid_list = [];
        $count = 0;
        while ($count < $step && $d = $redis->lPop($rkey)){
            array_push($openid_list, $d);
            $count++;
        }
        echo ('执行' . $count . "次<br>" . PHP_EOL);
        $coroutine->send([
            'module' => 'test',
            'list' => $openid_list,
            'mp_info' => $mp_info
        ]);
        if($redis->lLen($rkey)){
            gc_mem_caches(); //清理缓存碎片
            echo "剩余任务量：" . $redis->lLen($rkey) . PHP_EOL;
            swoole_timer_after($interval, [$this, 'afterCallback'], $params); //持续
        }else{
            //$redis->del('taskon'.$task['id']);
            echo "完成" . PHP_EOL;
        }
    }

    /**
     * 执行任务
     * @param $params
     * @return array|bool|false|mixed|\PDOStatement|string|\think\Model
     * @throws \think\exception\DbException
     * @throws \think\Exception
     */
    public function task($params = []){
        $job = $params['job'];
        $task_id = $params['task_id'];
        $task = $this->taskM->getOneJoin([
            'alias' => 'task',
            'join' => [
                ['mp_templatemsg msg', 'msg.id=task.templatemsg_id']
            ],
            'where' => ['task.id' => $task_id, 'task.status' => ['in', [0, 1]], 'push_time' => ['<=', time()]],
            'field' => ['task.*', 'msg.template_id', 'msg.content', 'msg.url']
        ]);
        //Log::error('attempts_' . $task['id'] . ':'.$job->attempts());
        $redis = $this->getRedis();
        if(empty($task) || $job->attempts() > 10000){
            $job->delete();
            var_dump('任务不存在');
            return true;
        }

        $r_key = 'openidlist' . $task_id;
        $refresh = !$task['status'];

        $mp_info = model('common/mp')->getOne($task['mpid']);
        $app = controller('mp/mp', 'event')->getApp($mp_info);

        if($refresh){ //首次执行
            $redis->del('tempmsg_count_'.$task_id);
            $redis->del($r_key);
            $limit = $params['limit'] ?? 10000;
            $update = [
                'id' => $task['id'],
                'complete_time' => 0,
                'status' => 1
            ];
            $this->taskM->updateOne($update);

            //生产
            $next_openid = '';
            while (true) {
                if($redis->lLen($r_key) >= $limit){
                    break;
                }
                $res = $app->user->list($next_openid);
                if (!empty($res['count'])) {
                    $openid_list = $res['data']['openid'];
                    $redis->lPush($r_key, ...$openid_list);
                    $next_openid = $res['next_openid'];
                }else{
                    break;
                }
            }
        }

        $config = config('system.develop');
        $coroutine = null /*new Client([
            'port' => empty($config['swoole_port']) ? 9502 : $config['swoole_port'],
            'host' => empty($config['swoole_host']) ? '0.0.0.0' : $config['swoole_host']
        ])*/;
        echo('长度:' . $redis->lLen($r_key)) . PHP_EOL;
        $interval = $params['interval'] ?? 5000;
        $step = $params['step'] ?? 30;
        $params = ['config' => $config, 'mp_info' => $mp_info, 'redis'=>$redis, 'rkey' => $r_key,
            'task' => $task, 'coroutine' => $coroutine, 'interval' => $interval, 'step' => $step
        ];

        swoole_timer_after($interval, [$this, 'taskTimer'], $params);
        if($redis->lLen($r_key)){
            $job->release(1); //续约任务
        }else{
            $job->delete();
            $redis->del($r_key);
        }
    }

    public function taskTimer($params = []){
        $coroutine = $params['coroutine'];
        $mp_info = $params['mp_info'];
        $app = controller('mp/mp', 'event')->getApp($mp_info);
        $redis = $params['redis'];
        $rkey = $params['rkey'];
        $task = $params['task'];
        $step = $params['step'] ?? 30;
        $interval = $params['interval'] ?? 5000;
        $openid_list = [];
        $count = 0;

        //$api = $app->template_message;
        $api = $app->user;
        while ($count < $step && $d = $redis->lPop($rkey)){
            /* //测试后放开
             * $api->send([
                'touser' => $d,
                'template_id' => $task['template_id'],
                'url' => $task['url'],
                'data' => json_decode($task['content'], true),
                'client_msg_id' => 'task_' . $task['id']
            ]);*/
            var_dump($api->get($d));
            $count++;
            $redis->incr('tempmsg_count_'.$task['id']);
        }
        echo ('执行' . $count . "条<br>" . PHP_EOL);
        /*$coroutine->send([
            'module' => 'templatemsg',
            'msg' => [
                'template_id' => $task['template_id'],
                'url' => $task['url'],
                'data' => json_decode($task['content'], true),
                'client_msg_id' => 'task_' . $task['id']
            ],
            'mp_info' => $mp_info,
            'openid_list' => $openid_list
        ]);*/

        $update = [
            'id' => $task['id'],
            'count' => $redis->get('tempmsg_count_'.$task['id'])
        ];
        if($redis->lLen($rkey)){
            echo "剩余任务量：" . $redis->lLen($rkey) . PHP_EOL;
            swoole_timer_after($interval, [$this, __FUNCTION__], $params); //持续
        }else{
            $update['status'] = 2;
            $update['complete_time'] = time();
            $update['count'] = $task['total'];
            $redis->del('tempmsg_count_'.$task['id']);
            echo "完成" . PHP_EOL;
        }
        $this->taskM->updateOne($update);
        gc_mem_caches(); //清理缓存碎片
    }
}